package com.atguigu;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Properties;

/**
 * @author yhm
 * @create 2023-11-24 15:24
 */
public class DimApp extends BaseApp {

    public static void main(String[] args) {
        new DimApp().start(10001,4,"dim_app", Constant.TOPIC_DB);
    }


    // 数据的具体处理
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 具体的数据处理逻辑
        // 筛选出维度表的数据  写入到对应的hbase表格中
        // 1. 获取主流数据topic_db 进行清洗过滤
        // 清洗加转换  推荐使用flatMap
        SingleOutputStreamOperator<JSONObject> jsonObjStream = etl(stream);
        jsonObjStream.print("topic_db");

        // 2. FlinkCDC实时监控获取配置表数据(存放哪些表示维度表)
        DataStreamSource<String> tableProcessStream = readTableProcess(env);
        tableProcessStream.print("table_process");

        // 3. 在hbase中创建出与之对应的dim_维度表

        // 4. 连接主流和配置表流

        // 5. 筛选出维度表数据

        // 6. 写出到hbase
    }

    private DataStreamSource<String> readTableProcess(StreamExecutionEnvironment env) {
        Properties props = new Properties();
        props.setProperty("useSSL", "false");
        props.setProperty("allowPublicKeyRetrieval", "true");

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOST)
                .port(Constant.MYSQL_PORT)
                .username(Constant.MYSQL_USER_NAME)
                .password(Constant.MYSQL_PASSWORD)
                .databaseList("gmall2023_config")
                .tableList("gmall2023_config.table_process_dim")
                .jdbcProperties(props)
                // 添加功能的参数
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();
        return  env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "table_process");
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    String type = jsonObject.getString("type");
                    JSONObject data = jsonObject.getJSONObject("data");
                    JSONObject old = jsonObject.getJSONObject("old");
                    if (!("bootstrap-start".equals(type) || "bootstrap-complete".equals(type))) {
                        // type正确的数据
                        if (data != null) {
                            out.collect(jsonObject);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }


}
